package com.mj.broadcast;

import com.mj.utils.Util;
import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.common.message.MessageExt;
import org.apache.rocketmq.remoting.protocol.heartbeat.MessageModel;

import java.util.List;

/**
 * 广播消息消费者
 */
public class BroadcastConsumer {
    public static void main(String[] args) throws MQClientException {
        // 1.和生产者一样，创建一个DefaultMQPushConsumer实例
        DefaultMQPushConsumer consumer = new DefaultMQPushConsumer(Util.ConsumerGroup);
        // 2.设置NameServer的地址，消费者通过这个地址与NameServer进行通信，来获取Broker的地址信息
        consumer.setNamesrvAddr(Util.NamesrvAddr);
        // 订阅一个或多个Topic，以及Tag来过滤需要消费的消息。我们订阅了"MyTopic1"，使用"*"来匹配此Topic下的所有Tag
        consumer.subscribe(Util.Topic, "*");
        //设置为广播消息消费者，所有的ConsumerGroup里的消费者同时进行消费（相当于ConsumerGroup不起作用了），但是同一个消息还是只能被1个consumer消费
        consumer.setMessageModel(MessageModel.BROADCASTING);
        // 3.注册消息监听器，用于处理从Broker接收到的消息。使用MessageListenerConcurrently接口的实现，表示并行消费
        consumer.registerMessageListener(new MessageListenerConcurrently() {
            @Override
            // 4.当收到消息时，方法会被调用。
            public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs, ConsumeConcurrentlyContext consumeConcurrentlyContext) {
                // 5.遍历消息列表，并打印每条消息的内容（注意：这里直接打印msg对象不会得到预期的消息内容字符串）
                for (MessageExt msg : msgs) {
                    // 所以我们打印msg.getBody()的内容，为了保留消息原样
                    System.out.println("已收到消息" + msg);
                }
                // 6.返回消费状态，这里表示消息已成功消费
                return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
            }
        });
        // 7.启动消费者实例
        consumer.start();
        // 8.打印日志消费者已启动
        System.out.println("消费者已启动");

    }
}
